package com.atguigu.edu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.common.GmallConfig;
import com.atguigu.edu.util.DruidPhoenixDSUtil;
import com.atguigu.edu.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class MyPhoenixSink extends RichSinkFunction<JSONObject> {

    private DruidDataSource druidDataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 创建连接池
        druidDataSource = DruidPhoenixDSUtil.getDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // 构建插入数据的sql
        String sql = createUpsertSQL(value);
        System.out.println(sql);

        DruidPooledConnection conn = null;
        try {
            conn = druidDataSource.getConnection();
        } catch (SQLException sqlException) {
            sqlException.printStackTrace();
            System.out.println("从 Druid 连接池获取连接对象异常");
        }
        PhoenixUtil.executeSQL(sql, conn);
    }

    private String createUpsertSQL(JSONObject value) {
        // upsert into tableName (columns) values(值)
        StringBuilder upsertSQL = new StringBuilder();
        Set<String> columns = value.getJSONObject("data").keySet();
        Collection<Object> values = value.getJSONObject("data").values();
        String sinkTable = value.getString("sinkTable");
        upsertSQL.append("upsert into ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("(")
                .append(StringUtils.join(columns,","))
                .append(") values ('")
                .append(StringUtils.join(values,"','"))
                .append("')");
        return upsertSQL.toString();

    }
}
